package com.cetcs.kmga.dataManager.util;


import com.google.common.collect.Lists;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.PageFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.stream.Collectors;


/**
 * Hbase数据库连接池以及通用的分页查询
 *
 * @author xutao
 * @version V1.0 创建时间：2017-08-30 11:27
 * Copyright 2017 by CETC
 */
public class HBasePool {

    private final static Logger LOGGER = LoggerFactory.getLogger(HBasePool.class);

    private final static String INDEX_NAME = "KMGA:";

    private static Configuration configuration;

    private static Connection connection = null;
    /**
     * 固定长度的线程池
     */
    private final ExecutorService FIXED_LENGTH_POOL = new ThreadPoolExecutor(10, 200,
            60*1000*10, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<Runnable>());;

    private final ExecutorService FIXED_ALL_LENGTH_POOL = Executors.newFixedThreadPool(10);

    static {
        configuration = HBaseConfiguration.create();
        configuration.addResource(new Path("hbase-site.xml"));
        configuration.addResource(new Path("core-site.xml"));
        configuration.set("hbase.client.ipc.pool.size", "10");
    }

    private static HBasePool instance = null;

    public static HBasePool getInstance() {
        if (instance == null) {
            instance = new HBasePool();
        }
        return instance;
    }

    public Connection getConnection() {
        try {
            if (connection == null) {
                connection = ConnectionFactory.createConnection(configuration);
            }
        } catch (IOException ex) {
            ex.printStackTrace();

        }
        return connection;
    }

    /**
     * 分页检索表数据。<br>
     * （如果在创建表时为此表指定了非默认的命名空间，则需拼写上命名空间名称，格式为【namespace:tablename】）。
     *
     * @param tableName   表名称(*)。
     * @param columnArr   需要查询的字段数组。
     * @param startRowKey 起始行键(可以为空，如果为空，则从表中第一行开始检索)。
     * @param endRowKey   结束行键(可以为空)。
     * @param filterList  检索条件过滤器集合(不包含分页过滤器；可以为空)。
     * @param maxVersions 指定最大版本数【如果为最大整数值，则检索所有版本；如果为最小整数值，则检索最新版本；否则只检索指定的版本数】。
     * @param pageModel   分页模型(*)。
     * @return 返回HBasePageModel分页对象。
     */
    public HBasePageModel scanResultByPageFilter(String tableName, List<String> columnArr, byte[] startRowKey, byte[] endRowKey, FilterList filterList, int maxVersions, HBasePageModel pageModel) {
        if (pageModel == null) {
            pageModel = new HBasePageModel(10);
        }
        if (maxVersions <= 0) {
            //默认只检索数据的最新版本
            maxVersions = Integer.MIN_VALUE;
        }
        pageModel.initStartTime();
        pageModel.initEndTime();
        if (StringUtils.isBlank(tableName)) {
            return pageModel;
        }
        Table table = null;
        Connection connection = getConnection();
        if (connection == null) {
            return pageModel;
        }
        try {
            table = connection.getTable(TableName.valueOf(tableName), FIXED_ALL_LENGTH_POOL);
            int tempPageSize = pageModel.getPageSize();
            boolean isEmptyStartRowKey = false;
            if (startRowKey == null) {
                //则读取表的第一行记录
                Result firstResult = selectFirstResultRow(tableName, filterList);
                if (firstResult.isEmpty()) {
                    return pageModel;
                }
                startRowKey = firstResult.getRow();
            }
            if (pageModel.getPageStartRowKey() == null) {
                isEmptyStartRowKey = true;
                pageModel.setPageStartRowKey(startRowKey);
            } else {
                if (pageModel.getPageEndRowKey() != null) {
                    pageModel.setPageStartRowKey(pageModel.getPageEndRowKey());
                }
                //从第二页开始，每次都多取一条记录，因为第一条记录是要删除的。
                tempPageSize += 1;
            }

            Scan scan = new Scan();
            scan.setStartRow(pageModel.getPageStartRowKey());
            if (endRowKey != null) {
                scan.setStopRow(endRowKey);
            }
            PageFilter pageFilter = new PageFilter(pageModel.getPageSize() + 1);
            if (filterList != null) {
                filterList.addFilter(pageFilter);
                scan.setFilter(filterList);
            } else {
                scan.setFilter(pageFilter);
            }
            if (maxVersions == Integer.MAX_VALUE) {
                scan.setMaxVersions();
            } else if (maxVersions == Integer.MIN_VALUE) {

            } else {
                scan.setMaxVersions(maxVersions);
            }
            scan.setCaching(1000);
            ResultScanner scanner = table.getScanner(scan);
            List<Result> resultList = Lists.newArrayList();
            List<List<String>> dataList = Lists.newArrayList();
            int index = 0;
            for (Result rs : scanner.next(tempPageSize)) {
                if (isEmptyStartRowKey == false && index == 0) {
                    index += 1;
                    continue;
                }
                if (!rs.isEmpty()) {
                    if (rs.getRow() != null) {
                        resultList.add(rs);
                        List<String> singleList = result2List(rs, columnArr);
                        if (singleList != null) {
                            dataList.add(singleList);
                        }
                    }
                }
                index += 1;
            }
            scanner.close();
            pageModel.setResultList(resultList);
            pageModel.setDataList(dataList);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                if (table != null) {
                    table.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        int pageIndex = pageModel.getPageIndex() + 1;
        pageModel.setPageIndex(pageIndex);
        if (pageModel.getResultList().size() > 0) {
            //获取本次分页数据首行和末行的行键信息
            byte[] pageStartRowKey = pageModel.getResultList().get(0).getRow();
            byte[] pageEndRowKey = pageModel.getResultList().get(pageModel.getResultList().size() - 1).getRow();
            pageModel.setPageStartRowKey(pageStartRowKey);
            pageModel.setPageEndRowKey(pageEndRowKey);
        }
        int queryTotalCount = pageModel.getQueryTotalCount() + pageModel.getResultList().size();
        pageModel.setQueryTotalCount(queryTotalCount);
        pageModel.initEndTime();
        pageModel.printTimeInfo();
        return pageModel;
    }

    public Map<String, Object> result2Map(Result result) {
        Map<String, Object> ret = new HashMap<>();
        if (result != null && result.listCells() != null) {
            String rowKey = "";
            for (Cell cell : result.listCells()) {
                String key = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                ret.put(key, value);
                rowKey = Bytes.toString(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());
            }

            ret.put("ID", rowKey);
        }
        return ret;
    }

    public Map<String, Object> findDataDetailByRowKey(String tbName, String rowKey) {
        Table table = null;
        Map<String, Object> detailMap = null;
        Connection conn = getConnection();
        if (conn == null) {
            return null;
        }
        try {
            table = conn.getTable(TableName.valueOf(tbName));
            Get get = new Get(Bytes.toBytes(rowKey));
            Result result = table.get(get);
            if (result != null) {
                detailMap = result2Map(result);
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                if (table != null) {
                    table.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }

        }
        return detailMap;
    }

    /**
     * 将result转换成List数组
     *
     * @param result    Hbase查询的结果集
     * @param columnArr 字段数组对象
     * @return
     */
    public List<String> result2List(Result result, List<String> columnArr) {
        List<String> ret = Lists.newArrayList();
        if (result == null) {
            return null;
        }
        String rowKey = new String(result.getRow());
        ret.add(rowKey);
        if (result != null && result.listCells() != null && columnArr != null && columnArr.size() > 0) {
            for (int i = 1; i < columnArr.size(); i++) {
                int index = i;
                //找出与字段值相等的数据
                List<String> singleList = result.listCells().stream().filter(p -> columnArr.get(index).equalsIgnoreCase(Bytes.toString(CellUtil.cloneQualifier(p)))).map(p ->
                        Bytes.toString(CellUtil.cloneValue(p))).collect(Collectors.toList());
                if (singleList != null && singleList.size() > 0) {
                    ret.add(singleList.get(0));
                } else {
                    ret.add(null);
                }
            }
        }
        return ret;
    }


    /**
     * 检索指定表的第一行记录。<br>
     * （如果在创建表时为此表指定了非默认的命名空间，则需拼写上命名空间名称，格式为【namespace:tablename】）。
     *
     * @param tableName  表名称(*)。
     * @param filterList 过滤器集合，可以为null。
     * @return
     */

    public Result selectFirstResultRow(String tableName, FilterList filterList) {
        if (StringUtils.isBlank(tableName)) return null;
        Table table = null;
        Connection conn = getConnection();
        if (conn == null) {
            return null;
        }
        try {
            table = conn.getTable(TableName.valueOf(tableName));
            Scan scan = new Scan();
            if (filterList != null) {
                scan.setFilter(filterList);
            }
            ResultScanner scanner = table.getScanner(scan);
            Iterator<Result> iterator = scanner.iterator();
            int index = 0;
            while (iterator.hasNext()) {
                Result rs = iterator.next();
                if (index == 0) {
                    scanner.close();
                    return rs;
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                if (table != null) {
                    table.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return null;
    }

    /**
     * 采用多线程的方式批量获取数据
     *
     * @param tbName  表名
     * @param rowKeys rowKey的数据
     * @return
     */
    public List<List<String>> findDataFromDbByRowKeys(String tbName, List<String> rowKeys, List<String> cloumnArr) throws ExecutionException, InterruptedException {
        if (rowKeys == null || rowKeys.size() <= 0) {
            return null;
        }
        //一次性查询rowKey的最大长度
        final int maxRowKeySize = 1000;
        int loopSize = rowKeys.size() % maxRowKeySize == 0 ? rowKeys.size()
                / maxRowKeySize : rowKeys.size() / maxRowKeySize + 1;
        List<Future<Result[]>> futureList = Lists.newArrayList();
        //循环加载数据
        for (int loop = 0; loop < loopSize; loop++) {
            int end = (loop + 1) * maxRowKeySize > rowKeys.size() ? rowKeys
                    .size() : (loop + 1) * maxRowKeySize;
            List<String> partRowKeys = rowKeys.subList(loop * maxRowKeySize,
                    end);
            HBaseDataGetter hbaseDataGetter = new HBaseDataGetter(tbName, partRowKeys);
            Future<Result[]> result = FIXED_LENGTH_POOL.submit(hbaseDataGetter);
            futureList.add(result);
        }
        List<List<String>> dataQueue = Lists.newArrayList();
        try {
            for (Future<Result[]> futureSingle : futureList) {
                Result[] rdArr = futureSingle.get();
                for (Result rt : rdArr) {
                    if (rt != null && !rt.isEmpty()) {
                        List<String> singleRet = result2List(rt, cloumnArr);
                        dataQueue.add(singleRet);
                    }
                }
            }
        } catch (Exception ex) {
            LOGGER.error("加载线程池满：" + ex.getMessage());
        }
        return dataQueue;
    }

    private void closeConn(Connection conn) throws IOException {
        if (conn != null) {
            conn.close();
            conn = null;
        }
    }
}
